package mylab.io.core;

import com.google.common.base.Preconditions;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import lombok.extern.slf4j.Slf4j;
import mylab.base.foundation.lang.Callback;
import mylab.io.base.BaseDas;
import mylab.io.base.model.Msg;
import mylab.io.base.model.Terminal;
import mylab.io.base.property.DasProperties;
import mylab.io.base.util.TerminalUtil;
import mylab.io.core.codec.Codec;
import mylab.io.core.context.CoreContext;
import mylab.io.core.util.NettyUtil;

/**
 * 自实现接入服务模板类，方便扩展实现各类通讯协议
 */
@Slf4j
public abstract class AbstractDas<P extends DasProperties, M extends Msg, C extends Codec<P, M, F>, F>
        extends BaseDas<CoreContext<P, M>, P, M> {

    protected C codec;
    protected Channel channel;
    protected ChannelFuture startFuture;

    public AbstractDas(CoreContext<P, M> context, C codec) {
        super(context);

        Preconditions.checkNotNull(codec, "codec is null");
        this.codec = codec;
    }

    protected abstract ChannelFuture actualStart() throws Exception;

    @Override
    protected void startFunc(Callback<Void> callback) {
        try {
            startFuture = actualStart();
            startFuture.addListener(f -> {
                if (!f.isSuccess()) {
                    //如果启动不成功，则关闭通道
                    startFuture.channel().close();

                    if (f.cause() != null) {
                        f.cause().printStackTrace();
                    }
                }
                //close时，异步执行 shutdownGracefully
                startFuture.channel().closeFuture().addListener((ChannelFutureListener) future -> {
                    future.channel().eventLoop().parent().shutdownGracefully();
                });
            });

            NettyUtil.dealFuture(startFuture, callback);
        } catch (Exception e) {
            Callback.dealOnException(callback, e);
        }
    }

    @Override
    protected void stopFunc(Callback<Void> callback) {
        NettyUtil.closeChannel(channel);
        Callback.dealOnSuccess(callback);
    }

    @Override
    protected void asyncSendFunc(M request, Callback<Void> callback) {
        try {
            //查找对应channel
            Channel channel = context.getChannel(request.getTo());
            Preconditions.checkNotNull(channel, "channel not found");
            Preconditions.checkState(channel.isOpen(), "channel is not open");
            Preconditions.checkState(channel.isActive(), "channel is not active");
            Preconditions.checkState(channel.isWritable(), "channle is not writeable");
            ChannelFuture future = channel.writeAndFlush(request);
            NettyUtil.dealFuture(future, callback);
        } catch (Exception e) {
            Callback.dealOnException(callback, e);
        }
    }

    @Override
    protected void checkBeforeSend(M msg) {
        super.checkBeforeSend(msg);

        Terminal remote;
        //若待发送消息无明确指定目标地址
        if (msg.getTo() == null && getContext().getDasPropertites().getCommon().isClientMode()) {
            String dasId = getId();
            remote = getContext().getDasCache().getFirstTerminalByDas(dasId);
            if (TerminalUtil.isValid(remote)) {
                //当有且仅有唯一 remote terminal 时，作为首选
                msg.setTo(remote);
            }
        }

        remote = msg.getTo();
        Preconditions.checkState(TerminalUtil.isValid(remote), "msg 目标地址未知");
    }


}
